package com.by.rpc.netty;

import com.by.log.Logger;
import com.by.rpc.provider.ServiceProvider;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import com.by.log.LoggerFactory;

/**
 * Netty RPC服务器，用于接收和处理RPC请求
 */
public class RpcServer {
    // 使用统一的日志框架
    private static final Logger log = LoggerFactory.getLogger(RpcServer.class);
    
    private final int port;
    private final ServiceProvider serviceProvider;
    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;
    private Channel channel;

    public RpcServer(int port, ServiceProvider serviceProvider) {
        this.port = port;
        this.serviceProvider = serviceProvider;
        this.bossGroup = new NioEventLoopGroup(1);
        this.workerGroup = new NioEventLoopGroup();
    }

    /**
     * 启动RPC服务器
     */
    public void start() {
        try {
            System.out.println("Starting RPC server on port: " + port);
            
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            // 长度字段处理器（保持不变，解决粘包问题）
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                            pipeline.addLast(new LengthFieldPrepender(4));

                            // 移除字符串编解码器，添加Kryo编解码器
                            pipeline.addLast(new KryoDecoder());  // 新增Kryo解码器
                            pipeline.addLast(new KryoEncoder());  // 新增Kryo编码器

                            // 自定义RPC服务端处理器（后续会修改）
                            pipeline.addLast(new RpcServerHandler(serviceProvider));
                        }
                    });

            // 绑定端口并启动服务器
            ChannelFuture future = bootstrap.bind(port).sync();
            channel = future.channel();
            log.info("RPC server started on port: {}", port);

            // 监听服务器通道关闭事件
            // future.channel().closeFuture().sync(); // 注释掉这行，避免阻塞
        } catch (InterruptedException e) {
            log.error("Failed to start RPC server on port: {}, error: {}", port, e.getMessage(), e);
            Thread.currentThread().interrupt();
        }
    }

    public void close() {
        log.info("Closing RPC server on port: {}", port);
        if (channel != null) {
            channel.close();
        }
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
        log.info("RPC server closed on port: {}", port);
    }
}